package cn.smileyan.demos.operators;


import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 例子 5：reduce 例子
 * @author smileyan
 */
public class WindowDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<Integer, String>> fixedInput = env.fromElements(
                Tuple2.of(1, "apple"),
                Tuple2.of(2, "banana"),
                Tuple2.of(1, "cherry"),
                Tuple2.of(2, "elderberry"),
                Tuple2.of(1, "firefly"),
                Tuple2.of(2, "giraffe"),
                Tuple2.of(2, "hippopotamus")
        );

        SingleOutputStreamOperator<Tuple2<Integer, String>> reduced = fixedInput
                .keyBy(v -> v.f0)
                .countWindow(3)
                .reduce((a, b) -> Tuple2.of(a.f0, a.f1 + ";" + b.f1))
                .name("hello window");

        reduced.print();

        env.execute("Flink KeyBy Example");
    }
}
